package com.levi.consumer;

import com.alibaba.fastjson.JSONObject;
import com.levi.pojo.Orders;
import com.levi.service.PointService;
import com.levi.vo.OrderDto;
import com.levi.vo.PointDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author shentong
 * @since 2022/6/9 9:15 PM
 */
@Component
@Slf4j
public class ConsumerListener
        implements MessageListenerConcurrently {

    @Autowired
    private PointService pointService;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt messageExt : list) {
            boolean result = consumerMsg(messageExt);
            if (result) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            } else return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        return null;
    }

    private boolean consumerMsg(MessageExt messageExt) {
        String msg = new String(messageExt.getBody());
        Orders orders = JSONObject.parseObject(msg, Orders.class);
        try {
            PointDto pointDto = new PointDto();
            pointDto.setOrderNo(orders.getId());
            pointDto.setPoints(10);
            pointDto.setUserId(1);
            pointService.addPoint(pointDto);
        } catch (Exception e) {
            e.printStackTrace();
            if (messageExt.getReconsumeTimes() > 3) {
                //TODO  发送短信或者邮件进行告警
                log.error("订单号==>{} 消费3次都不成功",orders.getId());
            }
        }
        return true;
    }
}
